package com.ruoyi.flink.util.read;

import com.alibaba.fastjson.JSONArray;
import com.ruoyi.flink.util.GetBeanFromResultUtil;
import com.ruoyi.flink.util.db.ClickHouseLink;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;

/**
 * @Description: 从ClickHouse中读取数据 (在实际项目中还是按一般的写法来吧, 这种写法可能会导致线程不安全)
 * @ClassName: ClickHoustRead
 * @author syc
 * @param <T>
 * @date: 2022年5月26日 下午7:38:40
 */
public class ClickHouseRead extends RichSourceFunction<JSONArray> {

	private static final long serialVersionUID = 1L;
	Connection connection = null;
    PreparedStatement preparedStatement = null;
    String sql;
    String url = null;
    String user = null;
    String password = null;
    public ClickHouseRead(String sql) {
        this.sql = sql;
    }
    
    public ClickHouseRead(String sql, String url, String user, String password) {
        this.sql = sql;
        this.url = url;
        this.user = user;
        this.password = password;
    }
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        if(url != null && user != null && password != null) {
        	connection = ClickHouseLink.getConn(url, user, password); 	
        } else {
        	connection = ClickHouseLink.getConn();
        }       
    }

    @Override
    public void run(SourceContext<JSONArray> sourceContext) throws Exception {
        preparedStatement = connection.prepareStatement(sql);
        ResultSet resultSet = preparedStatement.executeQuery();
        if(resultSet!=null) {
            while(resultSet.next()) {
                sourceContext.collect(GetBeanFromResultUtil.resultSetToJsonArry(resultSet));
            }
        }
    }

    @Override
    public void cancel() {
        try {
            if(preparedStatement != null) {
                preparedStatement.close();
            }
            if (connection != null) {
                connection.close();
            }
        } catch (SQLException throwables) {
            throwables.printStackTrace();
        }

    }
}
